package com.atguigu.gmall.realtime.handler

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import com.atguigu.gmall.realtime.handler.JoinHandler.JoinType.JoinType
import com.atguigu.gmall.realtime.util.{Json4sUtil, RedisUtil}
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

import java.util
import scala.collection.mutable.ListBuffer
import scala.reflect.{ClassTag, classTag}
import scala.collection.JavaConverters._
import scala.runtime.Nothing$

object JoinHandler {


  def join[K: ClassTag, L:ClassTag, R:ClassTag   , M: ClassTag](leftDStream: DStream[(K, L)]
                                                               , rightDStram: DStream[(K, R)],
                                                               joinType: JoinType, expireTimeSec: Int, joinName: String
                                                               , matchFunc: (Option[L], Option[R]) => M
                                                              ): DStream[M] = {

    val joinLeftPrefix = s"JOIN_CACHE:$joinName:L"
    val joinRightPrefix = s"JOIN_CACHE:$joinName:R"


    val fullJoinStream: DStream[(K, (Option[L], Option[R]))] = leftDStream.fullOuterJoin(rightDStram)
    val matchStream: DStream[M] = fullJoinStream.flatMap { case (key, (leftOption, rightOption)) =>
      var hasLeft = false
      var hasRight = false
      val jedisClient: Jedis = RedisUtil.getJedisClient

      val matchList = ListBuffer[M]()
      if (leftOption != None) {
        val leftValue: L = leftOption.get
        hasLeft = true
        if (rightOption != None) {
          //左右都在
          hasRight = true
          val rightValue: R = rightOption.get
          val someL: Option[L] = Some(leftValue)
          val someR: Option[R] = Some(rightValue)
          val matchValue: M = matchFunc(someL, someR)
          matchList.append(matchValue)

        }

        //写缓存 key不为空
        //Set
        if (key != null && key != "0") {
          val leftKey = s"$joinLeftPrefix:$key"
          jedisClient.sadd(leftKey, JSON.toJSONString(leftValue, new SerializeConfig(true)));
          jedisClient.expire(leftKey, expireTimeSec);
          //读缓存
          val rightKey = s"$joinRightPrefix:$key"
          val rightJsonSet: util.Set[String] = jedisClient.smembers(rightKey)
          if (rightJsonSet.size() > 0) {
            hasRight = true
            for (rightJsonStr <- rightJsonSet.asScala) {
                val rightValue: R = JSON.parseObject[R](rightJsonStr, classTag[R].runtimeClass)

            //  val rightValue: R = Json4sUtil.strToCaseClass[R](rightJsonStr)
              val someL=Some(leftValue)
              val someR=Some(rightValue)
              val matchValue: M = matchFunc(someL, someR)
              matchList.append(matchValue)
            }
          }

        }

        //如果是左连接允许只有一边数据输出
        if (joinType == JoinType.Left && hasLeft && !hasRight) {
          val someL=Some(leftValue)
          val matchValue: M = matchFunc(someL,  None)
          matchList.append(matchValue)
        }

      } else {
        hasRight = true
        val rightValue: R = rightOption.get
        //写缓存
        //Set
        if (key != null && key != "0") {
     //     println(s"right::::$rightValue")
          val rightKey = s"$joinRightPrefix:$key"
          jedisClient.sadd(rightKey, JSON.toJSONString(rightValue, new SerializeConfig(true)));
          jedisClient.expire(rightKey, expireTimeSec);
          //读缓存
          val leftKey = s"$joinRightPrefix:$key"
          val leftJsonSet: util.Set[String] = jedisClient.smembers(leftKey)
          if (leftJsonSet.size() > 0) {
            hasLeft = true
            for (leftJsonStr <- leftJsonSet.asScala) {
               val leftValue: L = JSON.parseObject[L](leftJsonStr, classTag[L].runtimeClass)
             // val leftValue: L = Json4sUtil.strToCaseClass[L](leftJsonStr)
              if (leftValue == null) {
              //  println("left none: and right =" + rightKey)
              }
              val someL=Some(leftValue)
              val someR=Some(rightValue)
              val matchValue: M = matchFunc(someL, someR)
              matchList.append(matchValue)
            }
          }
        }
        //如果是右连接允许只有一边数据输出
        if (joinType == JoinType.Right && !hasLeft && hasRight) {
          val matchValue: M = matchFunc(None, Some(rightValue) )
          matchList.append(matchValue)
        }


      }
      jedisClient.close()
      matchList

    }
    matchStream

  }

  object JoinType extends Enumeration {
    type JoinType = Value
    val Left, Right, Inner = Value
  }


  def main(args: Array[String]): Unit = {
     test[String,String,String] {(a,b)=>
          "Aaa"
     }
  }

  def test[L:Manifest,R:Manifest,M:Manifest]( matchFunc:(Option[L], Option[R])=>M): Unit ={

    matchFunc(None,None)
  }

}
